package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/10/30
 * 将流中的数据同步到HBase表中
 */
public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection hBaseConn;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
        JSONObject jsonObj = tup2.f0;
        //获取对业务数据库维度表进行的操作类型
        String type = jsonObj.getString("type");
        //{"tm_name":"Redmi","id":1}
        jsonObj.remove("type");
        TableProcessDim tableProcessDim = tup2.f1;
        //获取要操作的HBase的表名
        String sinkTable = tableProcessDim.getSinkTable();
        //获取rowkey<<的值>>
        String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());

        //根据对业务数据库维度表进行的操作的类型 决定对HBase的维度表进行什么样的操作
        if("delete".equals(type)){
            //说明从业务数据库维度表中删除了一条数据  那么对应的需要从HBase表中也将这条数据删除掉
            HBaseUtil.deleteRow(hBaseConn, Constant.HBASE_NAMESPACE,sinkTable,rowKey);
        }else{
            //insert、update、bootstrap-insert    对HBase表执行put操作
            String sinkFamily = tableProcessDim.getSinkFamily();
            HBaseUtil.putRow(hBaseConn,Constant.HBASE_NAMESPACE,sinkTable,rowKey,sinkFamily,jsonObj);
        }

        //如果维度数据发生了变化，清除Redis中缓存的数据
        if("delete".equals(type)||"update".equals(type)){
            System.out.println("从Redis中删除"+sinkTable+"表的"+rowKey+"数据");
            jedis.del(RedisUtil.getKey(sinkTable,rowKey));
        }
    }
}
